package cn.ac.iscas.service.task.flink;

import ch.qos.logback.classic.Logger;
import cn.ac.iscas.base.utils.SpringUtil;
import cn.ac.iscas.config.MiddlePlatformConfig;
import cn.ac.iscas.domain.NodeTask;
import cn.ac.iscas.domain.source.FlinkSourceColumn;
import cn.ac.iscas.service.component.ComponentService;
import cn.ac.iscas.service.queue.SqlExecuteQueueService;
import cn.ac.iscas.service.queue.flink.FlinkTableQueueService;
import cn.ac.iscas.service.sink.MySqlSink;
import cn.ac.iscas.service.source.CsvSource;
import cn.ac.iscas.service.source.MySqlSource;
import cn.ac.iscas.service.source.RocketMqSource;
import cn.ac.iscas.service.transformer.*;

import cn.ac.iscas.service.transformer.statistics.StatisticsTransformer;
import org.springframework.stereotype.Service;

import java.util.*;

/**
 * @author LJian
 * @version 1.0
 * @description: TODO
 * @date 2022/5/25 9:48
 */
@Service
public class FlinkSqlTask extends NodeTask{
    private String nodeType;
    private String kind;
    private String taskId;
    private Map<String, Object> parameters ;

    public FlinkSqlTask() {

    }

    /** 
     * @description:  
     * @param: builder 
     * @return:  
     * @author LJian
     * @date: 2022/12/27 9:09
     */ 
    private FlinkSqlTask(Builder builder) {
        super(builder.jobId, builder.taskId, builder.dependences);
        this.kind = builder.kind;
        this.nodeType = builder.nodeType;
        this.parameters = builder.parameters;
    }

    /** 
     * @description:  
     * @param: null 
     * @return:  
     * @author LJian
     * @date: 2022/12/27 9:00
     */ 
    public static class Builder{
        private String jobId;
        private String taskId;
        private Set<String> dependences;
        private String nodeType;
        private String kind;
        private Map<String, Object> parameters;

        public Builder(String jobId, String taskId, Set<String> dependences) {
            this.jobId = jobId;
            this.taskId = taskId;
            this.dependences = dependences;
        }

        public Builder jobId(String jobId) {
            this.jobId = jobId ;
            return this;
        }

        public Builder dependences(Set<String> dependences) {
            this.dependences = dependences ;
            return this ;
        }

        public Builder nodeType(String nodeType) {
            this.nodeType = nodeType ;
            return this;
        }

        public Builder kind(String kind) {
            this.kind = kind ;
            return this;
        }

        public Builder parameters(Map<String, Object> parameters) {
            this.parameters = parameters ;
            return this;
        }

        public FlinkSqlTask build() {
            return new FlinkSqlTask(this);
        }

    }


    /**
     * @description:  
     * @param: nodeType 
     * @param: kind 
     * @param: parameters 
     * @param: jobId 
     * @param: taskId 
     * @param: dependences 
     * @return:  
     * @author LJian
     * @date: 2022/12/27 9:09
     */ 
    public FlinkSqlTask(String nodeType, String kind, Map<String, Object> parameters, String jobId, String taskId, Set<String> dependences) {
        super(jobId, taskId, dependences);

        this.parameters = parameters;
        this.nodeType = nodeType;
        this.kind = kind;

    }

    @Override
    public Map<String, Object> doTask() throws Exception {
        Logger logger = getLogger().getLogger(getJobId());
        logger.info("do task: {}", getTaskId());
        SqlExecuteQueueService sqlExecuteQueueService = SpringUtil.getBean(SqlExecuteQueueService.class);
        FlinkTableQueueService flinkTableQueueService = SpringUtil.getBean(FlinkTableQueueService.class);
        MiddlePlatformConfig middlePlatformConfig = SpringUtil.getBean(MiddlePlatformConfig.class);
        ComponentService componentService = SpringUtil.getBean(ComponentService.class);

        String frontTableName = "";
        List<FlinkSourceColumn> frontColumns = new ArrayList<>();
        if (getDependences() != null &&!getDependences().isEmpty()) {
            frontTableName = flinkTableQueueService.getTableOrViewName(getJobId(), (String) this.getDependences().toArray()[0]);
            frontColumns = flinkTableQueueService.getColumns(getJobId(), (String) this.getDependences().toArray()[0], frontTableName);
        }

        if (nodeType.equalsIgnoreCase("reader") || nodeType.equalsIgnoreCase("writer")) {
            //String dataModel = (String) parameters.getOrDefault("dataModel", "");
            if (kind.toLowerCase(Locale.ROOT).contains("mysql") || kind.toLowerCase(Locale.ROOT).contains("dameng") ) {
                //String dataModel = (String) parameters.getOrDefault("name", "");
                String dataModel = (String) parameters.getOrDefault("tableName", "");
                parameters.clear();
                parameters = componentService.getDataSourceByDataModel(dataModel);
                //根据数据模型获取的数据源只包含数据库级，数据表名=模型名称
                parameters.put("table", dataModel);
            }
        }

        if (nodeType.equalsIgnoreCase("reader")) {
            if (kind.toLowerCase(Locale.ROOT).contains("mysql")) {
                String mysqlCreateTableSql = new MySqlSource(this.getJobId(), this.getTaskId()).getSql(parameters);
                sqlExecuteQueueService.add(this.getJobId(), mysqlCreateTableSql);
                System.out.println("++++++++++++" + mysqlCreateTableSql);
            } else if (kind.toLowerCase(Locale.ROOT).contains("csv")) {
                String csvCreateTableSql = new CsvSource(this.getJobId(), this.getTaskId()).getSql(parameters);
                sqlExecuteQueueService.add(this.getJobId(), csvCreateTableSql);
                System.out.println("++++++++++++" + csvCreateTableSql);
            } else if (kind.toLowerCase(Locale.ROOT).contains("rocketmqreader")) {
                String rocketmqCreateTableSql = new RocketMqSource(this.getJobId(), this.getTaskId()).getSql(parameters);
                sqlExecuteQueueService.add(this.getJobId(), rocketmqCreateTableSql);
                System.out.println("++++++++++++" + rocketmqCreateTableSql);
            } else {
                System.out.println("不支持的数据源");
            }

            System.out.println();
        } else if (nodeType.equalsIgnoreCase("transformer")) {
            switch (kind.toLowerCase(Locale.ROOT)) {
                case "invalid_checker" :
                    String invalidCheckSql = new InvalidCheckerTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), invalidCheckSql);
                    break;
                case "add_constant_columns" :
                    String addConstantColumnsSql = new AddConstantColumnsTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), addConstantColumnsSql);
                    break;
                case "filter_condition" :
                    String filterConditionSql = new FilterConditionTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), filterConditionSql);
                    break;
                case "bucket" :
                    String bucketSql = new BucketTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), bucketSql);
                    break;
                case "select_column" :
                    String selectColumn = new SelectColumnsTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), selectColumn);
                    break;
                case "fullfil_null":
                case "null_fill_trans":
                    String fullfilNullSql = new FullfilNullTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), fullfilNullSql);
                    break;
                case "deduplication_trans":
                    List<String> deduplicationSql = new DeduplicationTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), deduplicationSql);
                    break;
                case "null_data_clean_trans":
                    String nullDataCleanSql = new CleanNullTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), nullDataCleanSql);
                    break;
                case "normalization_trans":
                    String normalizationSql = new NormalizationTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), normalizationSql);
                    break;
                case "boundary_trans":
                    String boundarySql = new BoundaryValueTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), boundarySql);
                    break;
                case "invalid_value_del_trans":
                    String invalidValueDelSql = new InvalidValueDeleteTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), invalidValueDelSql);
                    break;
                case "statistics_trans":
                    String statisticsSql = new StatisticsTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), statisticsSql);
                    break;
                case "togglecase_trans":
                    String togglecaseSql = new ToggleCaseTransformer(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), togglecaseSql);
                    break;
                default:
                    break;
            }
        } else if (nodeType.equalsIgnoreCase("writer")) {
                if (kind.toLowerCase(Locale.ROOT).contains("mysql")) {
                    List<String> mysqlInsertTableSql = new MySqlSink(this.getJobId(), this.getTaskId(), frontTableName, frontColumns).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), mysqlInsertTableSql);
                    System.out.println("++++++++++++" + mysqlInsertTableSql);
                } else if (kind.toLowerCase(Locale.ROOT).contains("csv")) {
                    String csvCreateTableSql = new CsvSource(this.getJobId(), this.getTaskId()).getSql(parameters);
                    sqlExecuteQueueService.add(this.getJobId(), csvCreateTableSql);
                    System.out.println("++++++++++++" + csvCreateTableSql);
                } else {
                    System.out.println("不支持的数据源");
                }

            System.out.println();
        }


        System.out.println("++++++++++++++++++++++" + this.getTaskId());

        String msg = "finish MyNodeTask( " + this.getTaskId() + " )";
        Map<String, Object> result = new HashMap<>();
        result.put("msg", msg);
        return result;
    }
}
